// Tencent is pleased to support the open source community by making
// 蓝鲸智云 - 监控平台/日志平台 (BlueKing - Monitor/Log) available.
// Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. All rights reserved.
// Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
// You may obtain a copy of the License at http://opensource.org/licenses/MIT
// Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
// an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
// specific language governing permissions and limitations under the License.
//

package controller

import (
	"context"
	"fmt"
	"path/filepath"
	"sync"
	"time"

	"github.com/TencentBlueKing/bkmonitor-kits/logger"
	_ "go.uber.org/automaxprocs"

	"github.com/TencentBlueKing/bk-collector/confengine"
	"github.com/TencentBlueKing/bk-collector/controller/pipeline"
	"github.com/TencentBlueKing/bk-collector/define"
	"github.com/TencentBlueKing/bk-collector/exporter"
	"github.com/TencentBlueKing/bk-collector/internal/monitoring"
	"github.com/TencentBlueKing/bk-collector/receiver"
)

const (
	ConfigFieldMaxProcs = "max_procs"
	ConfigFieldLogging  = "logging"
)

type Controller struct {
	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup

	receiver      *receiver.Receiver
	exporter      *exporter.Exporter
	originalTasks *define.TaskQueue
	derivedTasks  *define.TaskQueue
	pipelineMgr   *pipeline.Manager
}

func SetupCoreNum(conf *confengine.Config) {
	define.SetCoreNum(conf.UnpackIntWithDefault(ConfigFieldMaxProcs, 0))
}

func SetupLogger(conf *confengine.Config) error {
	type LogConfig struct {
		Stdout  bool   `config:"stdout"`
		Level   string `config:"level"`
		Format  string `config:"format"`
		Path    string `config:"path"`
		MaxSize int    `config:"maxsize"`
		MaxAge  int    `config:"maxage"`
		Backups int    `config:"backups"`
	}
	var logCfg LogConfig
	if err := conf.UnpackChild(ConfigFieldLogging, &logCfg); err != nil {
		return err
	}

	logger.SetOptions(logger.Options{
		Stdout:     logCfg.Stdout,
		Format:     logCfg.Format,
		Filename:   filepath.Join(logCfg.Path, "bk-collector.log"),
		MaxSize:    logCfg.MaxSize,
		MaxAge:     logCfg.MaxAge,
		MaxBackups: logCfg.Backups,
		Level:      logCfg.Level,
	})
	return nil
}

func New(conf *confengine.Config) (*Controller, error) {
	SetupCoreNum(conf)
	if err := SetupLogger(conf); err != nil {
		return nil, err
	}

	exporterMgr, err := exporter.New(conf)
	if err != nil {
		return nil, err
	}

	configMgr, err := pipeline.NewManager(conf)
	if err != nil {
		return nil, err
	}

	receiverMgr, err := receiver.New(conf)
	if err != nil {
		return nil, err
	}

	ctx, cancel := context.WithCancel(context.Background())
	return &Controller{
		ctx:           ctx,
		cancel:        cancel,
		receiver:      receiverMgr,
		exporter:      exporterMgr,
		pipelineMgr:   configMgr,
		originalTasks: define.NewTaskQueue(define.PushModeGuarantee),
		derivedTasks:  define.NewTaskQueue(define.PushModeGuarantee),
	}, nil
}

func (c *Controller) Start() error {
	for i := 0; i < define.CoreNums(); i++ {
		go c.consumeRecords(c.ctx)
		go c.dispatchOriginalTasks(c.ctx)
		go c.dispatchDerivedTasks(c.ctx)
	}

	if err := c.receiver.Start(); err != nil {
		return err
	}
	if err := c.exporter.Start(); err != nil {
		return err
	}
	return nil
}

func (c *Controller) Stop() error {
	if err := c.receiver.Stop(); err != nil {
		return err
	}
	if err := c.exporter.Stop(); err != nil {
		return err
	}

	c.cancel()
	c.wg.Wait()
	logger.Info("controller has already stopped")
	return nil
}

func (c *Controller) Reload(conf *confengine.Config) error {
	t0 := time.Now()
	logger.Info("receiver reload signal, start to reload")

	if err := c.pipelineMgr.Reload(conf); err != nil {
		monitoring.ControllerReloadErrorTotal.WithLabelValues().Inc()
		logger.Errorf("failed to reload pipeline manager: %v", err)
		return err
	}
	if err := c.receiver.Reload(conf); err != nil {
		monitoring.ControllerReloadErrorTotal.WithLabelValues().Inc()
		logger.Errorf("failed to reload receiver: %v", err)
		return err
	}

	logger.Infof("reload finished, take: %v", time.Since(t0))
	monitoring.ControllerReloadSuccessTotal.WithLabelValues().Inc()
	return nil
}

func (c *Controller) submitTasks(q *define.TaskQueue, record *define.Record, pipelines []pipeline.Pipeline) {
	for i := 0; i < len(pipelines); i++ {
		if i == 0 {
			q.Push(define.NewTask(record, pipelines[i].Name(), pipelines[i].Processors()))
			continue
		}
		// clone 成本较高 所以仅当有多条流水线的时候才需要 clone
		q.Push(define.NewTask(record.Clone(), pipelines[i].Name(), pipelines[i].Processors()))
	}
}

func (c *Controller) consumeRecords(ctx context.Context) {
	c.wg.Add(1)
	defer c.wg.Done()

	for {
		select {
		case record, ok := <-receiver.Records():
			if !ok {
				return
			}
			pipelines := c.pipelineMgr.GetPipelines(record.RecordType)
			c.submitTasks(c.originalTasks, record, pipelines)

		case <-ctx.Done():
			return
		}
	}
}

// dispatchOriginalTasks 分发原始任务
func (c *Controller) dispatchOriginalTasks(ctx context.Context) {
	c.wg.Add(1)
	defer c.wg.Done()

loop:
	for {
		select {
		case task, ok := <-c.originalTasks.Get():
			if !ok {
				return
			}

			start := time.Now()
			rtype := task.Record().RecordType
			var lvs []string
			for i := 0; i < task.StageCount(); i++ {
				dataId := fmt.Sprintf("%d", task.Record().Token.Get(rtype))
				lvs = []string{task.PipelineName(), string(rtype), dataId}

				// 任务执行应该事务的 一旦中间某一环执行失败那就整体失败
				stage := task.StageAt(i)
				logger.Debugf("process original stage: %s, recordType: %+v", stage, task.Record().RecordType)
				derivedRecord, err := c.pipelineMgr.GetProcessor(stage).Process(task.Record())
				if err == define.ErrSkipEmptyRecord {
					monitoring.PipelineSkippedTotal.WithLabelValues(append(lvs, stage)...).Inc()
					logger.Warnf("skip empty record '%s' at stage: %v, token: %+v, err: %v", task.Record().RecordType, stage, task.Record().Token, err)
					goto loop
				}

				if err != nil {
					logger.Errorf("failed to process task: %v", err)
					monitoring.PipelineDroppedTotal.WithLabelValues(append(lvs, stage)...).Inc()
					goto loop
				}

				if derivedRecord != nil {
					pipelines := c.pipelineMgr.GetPipelines(derivedRecord.RecordType)
					derivedRecord.Unwrap()
					c.submitTasks(c.derivedTasks, derivedRecord, pipelines)
				}
			}

			exporter.PublishRecord(task.Record())
			logger.Debugf("original handle record: %+v, token: %+v", task.Record().RecordType, task.Record().Token)
			monitoring.PipelineHandledTotal.WithLabelValues(lvs...).Inc()
			monitoring.PipelineHandledDuration.WithLabelValues(lvs...).Observe(float64(time.Since(start).Milliseconds()))

		case <-ctx.Done():
			return
		}
	}
}

// dispatchDerivedTasks 分发派生任务
func (c *Controller) dispatchDerivedTasks(ctx context.Context) {
	c.wg.Add(1)
	defer c.wg.Done()

loop:
	for {
		select {
		case task, ok := <-c.derivedTasks.Get():
			if !ok {
				return
			}

			start := time.Now()
			rtype := task.Record().RecordType
			var lvs []string
			for i := 0; i < task.StageCount(); i++ {
				dataId := fmt.Sprintf("%d", task.Record().Token.Get(rtype))
				lvs = []string{task.PipelineName(), string(rtype), dataId}

				// 任务执行应该事务的 一旦中间某一环执行失败那就整体失败
				// 无需再关注是否为 derived 类型
				stage := task.StageAt(i)
				logger.Debugf("process derived stage: %s, recordType: %+v", stage, task.Record().RecordType)
				_, err := c.pipelineMgr.GetProcessor(stage).Process(task.Record())
				if err == define.ErrSkipEmptyRecord {
					logger.Warnf("skip empty record '%s' at stage: %v, token: %+v, err: %v", task.Record().RecordType, stage, task.Record().Token, err)
					monitoring.PipelineSkippedTotal.WithLabelValues(append(lvs, stage)...).Inc()
					goto loop
				}

				if err != nil {
					logger.Errorf("failed to process task: %v", err)
					monitoring.PipelineDroppedTotal.WithLabelValues(append(lvs, stage)...).Inc()
					goto loop
				}
			}

			exporter.PublishRecord(task.Record())
			logger.Debugf("derived handle record: %+v, token: %+v", task.Record().RecordType, task.Record().Token)
			monitoring.PipelineHandledTotal.WithLabelValues(lvs...).Inc()
			monitoring.PipelineHandledDuration.WithLabelValues(lvs...).Observe(float64(time.Since(start).Milliseconds()))

		case <-ctx.Done():
			return
		}
	}
}
